package com.mini.rpc.consumer;

import com.mini.rpc.common.*;
import com.mini.rpc.consumer.util.PropertiesUtil;
import com.mini.rpc.protocol.MiniRpcProtocol;
import com.mini.rpc.protocol.MsgHeader;
import com.mini.rpc.protocol.MsgType;
import com.mini.rpc.protocol.ProtocolConstants;
import com.mini.rpc.provider.registry.RegistryService;
import com.mini.rpc.serialization.SerializationTypeEnum;
import io.netty.channel.DefaultEventLoop;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
public class RpcInvokerProxy implements InvocationHandler {

    private final String serviceVersion;
    private final long timeout;
    private final RegistryService registryService;
    private final RpcProperties rpcProperties;
    private final String serializerAlgo;


    public RpcInvokerProxy(String serviceVersion, long timeout, RegistryService registryService) {
        this(serviceVersion, timeout, registryService, null);
    }

    public RpcInvokerProxy(String serviceVersion, long timeout, RegistryService registryService, RpcProperties rpcProperties) {
        this.serviceVersion = serviceVersion;
        this.timeout = timeout;
        this.registryService = registryService;
        this.rpcProperties = rpcProperties;
        this.serializerAlgo = PropertiesUtil.getValue("rpc.serializer.algorithm");
    }

    /**
     * 服务消费者实现协议编码, 向服务提供者发送调用数据
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 构造 RPC 协议对象
        MiniRpcProtocol<MiniRpcRequest> protocol = new MiniRpcProtocol<>();
        MsgHeader header = new MsgHeader();
        long requestId = MiniRpcRequestHolder.REQUEST_ID_GEN.incrementAndGet();
        header.setMagic(ProtocolConstants.MAGIC);
        header.setVersion(ProtocolConstants.VERSION);
        header.setRequestId(requestId);
        // 序列化算法 通过配置文件来管理
        // header.setSerialization((byte) SerializationTypeEnum.HESSIAN.getType());
        header.setSerialization((byte) SerializationTypeEnum.findByName(serializerAlgo).getType());
        header.setMsgType((byte) MsgType.REQUEST.getType());
        header.setStatus((byte) 0x1);
        protocol.setHeader(header);

        MiniRpcRequest request = new MiniRpcRequest();
        request.setServiceVersion(this.serviceVersion);
        request.setClassName(method.getDeclaringClass().getName());
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setParams(args);
        protocol.setBody(request);

        // todo 为什么是手动new出来一个呢? 而不是依赖注入方式
        RpcConsumer rpcConsumer = new RpcConsumer(rpcProperties);
        MiniRpcFuture<MiniRpcResponse> future = new MiniRpcFuture<>(new DefaultPromise<>(new DefaultEventLoop()), timeout);
        MiniRpcRequestHolder.REQUEST_MAP.put(requestId, future);
        // 发起 RPC 远程调用
        rpcConsumer.sendRequest(protocol, this.registryService);

        // TODO hold request by ThreadLocal

        // 等待 RPC 调用执行结果
        Object data = null;
        MiniRpcResponse rpcResponse;
        try {
            rpcResponse = future.getPromise().get(future.getTimeout(), TimeUnit.MILLISECONDS);
            data = rpcResponse.getData();
            if (data == null) {
                String message = rpcResponse.getMessage();
                log.error(">>> rpc invoke error: {}", message);
                throw new RuntimeException("invoke error: " + message);
            }
        } catch (TimeoutException e) {
            log.error(">>> rpc invoke Timeout: ", e);
            throw new RuntimeException(e);
        } catch (InterruptedException | ExecutionException e) {
            rpcResponse = future.getPromise().get(future.getTimeout(), TimeUnit.MILLISECONDS);
            log.error(">>>>>>>>> rpc invoke error: {}", rpcResponse.getMessage(), e);
        }
        return data;
    }
}
